[improve][api] Add Builder Methods to Create Message-based TableView #24809
Conversation
…raw Pulsar message
github-actions
bot
added
the
doc-required
…agement by adding retain() before storing message to rawMessages - Prevent premature message release that causes getRawMessage to return null keys. - Ensure message reference count is balanced by retaining message before storing and releasing old message after replacement. - Keep message release in finally block for safety.
There was a problem hiding this comment.
Thanks for the contribution!
Since this changes the Pulsar API, we'd need to first go through the PIP process to modify the public API.
Instead of adding a separate ConcurrentMap for storing the messages, I'd suggest going with an approach where the TableViewBuilder would have separate methods for creating a view for messages.
TableView<Message<T>> createForMessages() throws PulsarClientException;
CompletableFuture<TableView<Message<T>>> createForMessagesAsync() throws PulsarClientException;I'm not exactly sure if there's any obstacles with this approach, but it seems that it could be a better way forward so that the existing runtime behavior of TableView implementation wouldn't change (like it does with the current PR changes).
pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java
Outdated
Show resolved
Hide resolved
|
Hi @lhotari, Thanks for the review and the great suggestion about using the I completely agree that we should avoid impacting the runtime behavior for existing users. As you recommended, I've created a PIP for this API change, which you can find here: #24842 Let's continue the design discussion over on the PIP PR. |
@namest504 you seemed to ignore the suggestion in the comment. I'd suggest revisiting the PIP accordingly and renaming it. You can reply to the comment on the PIP PR, #24842 (review). |
Signed-off-by: namest504 <[email protected]>
Signed-off-by: namest504 <[email protected]>
Signed-off-by: namest504 <[email protected]>
namest504
changed the title
… method Signed-off-by: namest504 <[email protected]>
2 participants
Fixes: #24744
Main Issue: #24744
PIP: 445 #24842
Motivation
The current
TableViewAPI only exposes the deserialized message value, which limits access to essential message metadata like properties, event time, or the raw message object itself.This PR introduces a flexible, non-breaking mechanism to create value-mapped views over a topic. It replaces the initial proposal of adding a
getRawMessage()method, which would have performance implications for all users. Instead, it adds acreateMappedmethod to theTableViewBuilder, allowing users to transform a fullMessage<T>into any custom objectVthat suits their needs. This provides maximum flexibility, from accessing the full raw message to creating custom, memory-efficient objects.Modifications
createMappedandcreateMappedAsynctoTableViewBuilder: These new methods accept aFunction<Message<T>, V>mapper, empowering users to define their own mapping from a message to a value in theTableView.TableViewImpltoAbstractTableView: To avoid code duplication and support different value types, the core logic was moved to an abstract base class.MessageMapperTableView: A new implementation ofAbstractTableViewthat handles the logic for the user-defined mapper function.TableView: The standardTableViewnow extendsAbstractTableViewfor a payload-only view, ensuring backward compatibility.Verifying this change
Added new unit tests in
TableViewTest.java:testCreateMapped: Verifies custom mapping logic, including tombstone message handling (when the mapper returnsnull).testCreateMappedWithIdentityMapper: Verifies that usingFunction.identity()correctly creates aTableView<Message<T>>with the full message object as the value.All existing unit tests pass, ensuring no regressions.
Make sure that the change passes the CI checks.
Does this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes
Documentation
docdoc-requireddoc-not-neededdoc-completeMatching PR in forked repository
PR in forked repository: fork-repo